package com.yifeng.repo.flink.data.transport;


import com.yifeng.repo.flink.data.transport.streaming.doris.DorisSink;
import com.yifeng.repo.flink.data.transport.streaming.doris.DorisStreamLoad;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class FlinkKafka2Doris {
    //kafka address
    private static final String bootstrapServer = "192.168.253.135:6667,192.168.253.136:6667,192.168.253.137:6667";
    //kafka groupName
    private static final String groupName = "test_doris";
    //kafka topicName
    private static final String topicName = "tp_yf03_i";
    //doris ip port
    private static final String hostPort = "10.9.8.56:8030";
    //doris dbName
    private static final String dbName = "POS_DB";
    //doris tbName
    private static final String tbName = "TEST_BASE_STORE";
    //doris userName
    private static final String userName = "root";
    //doris password
    private static final String password = "";
    //doris columns
    private static final String columns = "ADDRESS_CODE,\n" +
            "COMPANY_CODE,\n" +
            "COMPANY_CODE4,\n" +
            "SALE_ORGAN,\n" +
            "DISTRIBUTION_CHANNEL,\n" +
            "PRODUCE_GROUP,\n" +
            "SALE_REGION,\n" +
            "SALE_OFFICE,\n" +
            "STORE_FULLNAME,\n" +
            "MEDICAL_INSURANCE_EXPLAIN,\n" +
            "PINYIN_CODE,\n" +
            "STORE_ABBREVIATION,\n" +
            "FILIALE_ADDRESS_CODE,\n" +
            "STORE_ADDRESS,\n" +
            "STORE_ADDRESS_2,\n" +
            "POSTALCODE,\n" +
            "CITY,\n" +
            "PROVINCE,\n" +
            "LINKAGE_PHONE,\n" +
            "CIRCLETYPE,\n" +
            "TOTALLENGTH_SHELF,\n" +
            "EXPRESS_STATION_,\n" +
            "DOES_MEDICARE,\n" +
            "MEDICARE_OPENTIME,\n" +
            "TOTALAREA,\n" +
            "FLOOR_AREA,\n" +
            "LONGITUDE,\n" +
            "IP,\n" +
            "IS_FLAGSHIP_SHOP,\n" +
            "REFER_SHOP,\n" +
            "STORE_STATUS,\n" +
            "STORE_AIR,\n" +
            "CREATOR,\n" +
            "CREATE_TIME,\n" +
            "MODIFIER,\n" +
            "MODIFY_TIME,\n" +
            "CREATOR_NAME,\n" +
            "MODIFIER_NAME,\n" +
            "LATITUDE,\n" +
            "RAW_SOURCE_CODE,\n" +
            "OPENING_DAY,\n" +
            "TARGET_SERVER,\n" +
            "TRAINING_BASE_TYPE,\n" +
            "IS_TRAINING_BASE,\n" +
            "OPEN_24H,\n" +
            "CITY_TYPE,\n" +
            "INTEREST,\n" +
            "BRANCH_CODE,\n" +
            "AREA_CODE,\n" +
            "SHOP_TYPE,\n" +
            "STORE_TYPE,\n" +
            "IS_CENTER,\n" +
            "IS_STARTUPSTORE,\n" +
            "SOURCE,\n" +
            "UNION_DIFFICULTY,\n" +
            "SALE_DIFFICULTY,\n" +
            "COMPETE_DIFFICULTY,\n" +
            "MOVE_DATE,\n" +
            "CLOSE_DATE,\n" +
            "COUNTY,\n" +
            "VILLAGE,\n" +
            "STORE_NUM,\n" +
            "USE_AREA,\n" +
            "MANAGE_AREA,\n" +
            "STORE_MAXIMUMDEPTH,\n" +
            "PRESCRIPTIONLENGTH_SHELF,\n" +
            "SELECTLENGTH_SHELF,\n" +
            "CHNMEDICINELENGTH_SHELF,\n" +
            "DOORHEAD_WIDTH,\n" +
            "ENDCAP_NUM,\n" +
            "IS_CORNER,\n" +
            "IS_GATHEREDPOINT,\n" +
            "RENTOUT_AREA,\n" +
            "BASE_TYPE,\n" +
            "SHOP_TYPEPREFIX,\n" +
            "SUPPLY_CENTER,\n" +
            "AREA_NAME,\n" +
            "BRANCH_NAME,\n" +
            "SHOP_FRONT_NUM,\n" +
            "CLOSE_DATE_PLAN,\n" +
            "VIRTUAL_COUNTY,\n" +
            "STORE_PROPERTIES,\n" +
            "PROPERTIES_DESCRIBE";
    //json format
    private static final String jsonFormat = "{\"$.ADDRESS_CODE\",\"$.COMPANY_CODE\",\"$.COMPANY_CODE4\",\"$.SALE_ORGAN\",\"$.DISTRIBUTION_CHANNEL\",\"$.PRODUCE_GROUP\",\"$.SALE_REGION\",\"$.SALE_OFFICE\",\"$.STORE_FULLNAME\",\"$.MEDICAL_INSURANCE_EXPLAIN\",\"$.PINYIN_CODE\",\"$.STORE_ABBREVIATION\",\"$.FILIALE_ADDRESS_CODE\",\"$.STORE_ADDRESS\",\"$.STORE_ADDRESS_2\",\"$.POSTALCODE\",\"$.CITY\",\"$.PROVINCE\",\"$.LINKAGE_PHONE\",\"$.CIRCLETYPE\",\"$.TOTALLENGTH_SHELF\",\"$.EXPRESS_STATION_\",\"$.DOES_MEDICARE\",\"$.MEDICARE_OPENTIME\",\"$.TOTALAREA\",\"$.FLOOR_AREA\",\"$.LONGITUDE\",\"$.IP\",\"$.IS_FLAGSHIP_SHOP\",\"$.REFER_SHOP\",\"$.STORE_STATUS\",\"$.STORE_AIR\",\"$.CREATOR\",\"$.CREATE_TIME\",\"$.MODIFIER\",\"$.MODIFY_TIME\",\"$.CREATOR_NAME\",\"$.MODIFIER_NAME\",\"$.LATITUDE\",\"$.RAW_SOURCE_CODE\",\"$.OPENING_DAY\",\"$.TARGET_SERVER\",\"$.TRAINING_BASE_TYPE\",\"$.IS_TRAINING_BASE\",\"$.OPEN_24H\",\"$.CITY_TYPE\",\"$.INTEREST\",\"$.BRANCH_CODE\",\"$.AREA_CODE\",\"$.SHOP_TYPE\",\"$.STORE_TYPE\",\"$.IS_CENTER\",\"$.IS_STARTUPSTORE\",\"$.SOURCE\",\"$.UNION_DIFFICULTY\",\"$.SALE_DIFFICULTY\",\"$.COMPETE_DIFFICULTY\",\"$.MOVE_DATE\",\"$.CLOSE_DATE\",\"$.COUNTY\",\"$.VILLAGE\",\"$.STORE_NUM\",\"$.USE_AREA\",\"$.MANAGE_AREA\",\"$.STORE_MAXIMUMDEPTH\",\"$.PRESCRIPTIONLENGTH_SHELF\",\"$.SELECTLENGTH_SHELF\",\"$.CHNMEDICINELENGTH_SHELF\",\"$.DOORHEAD_WIDTH\",\"$.ENDCAP_NUM\",\"$.IS_CORNER\",\"$.IS_GATHEREDPOINT\",\"$.RENTOUT_AREA\",\"$.BASE_TYPE\",\"$.SHOP_TYPEPREFIX\",\"$.SUPPLY_CENTER\",\"$.AREA_NAME\",\"$.BRANCH_NAME\",\"$.SHOP_FRONT_NUM\",\"$.CLOSE_DATE_PLAN\",\"$.VIRTUAL_COUNTY\",\"$.STORE_PROPERTIES\",\"$.PROPERTIES_DESCRIBE\"}";

    public static void main(String[] args) throws Exception {

        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServer);
        props.put("group.id", groupName);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "earliest");
        props.put("max.poll.records", "10000");

        StreamExecutionEnvironment blinkStreamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        blinkStreamEnv.enableCheckpointing(10000);
        blinkStreamEnv.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>(topicName,
                new SimpleStringSchema(),
                props);

        DataStreamSource<String> dataStreamSource = blinkStreamEnv.addSource(flinkKafkaConsumer);

        DorisStreamLoad dorisStreamLoad = new DorisStreamLoad(hostPort, dbName, tbName, userName, password);

        dataStreamSource.addSink(new DorisSink(dorisStreamLoad,columns,jsonFormat));

        blinkStreamEnv.execute("flink kafka to doris");

    }
}
